package com.shujia.flink.state;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class Demo04ValueStateWithCK {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         * CheckPoint会定期的将状态以及消费的偏移量（Kafka）一起保存到HDFS
         * Flink提供了两种状态保存的方式（StateBackend状态后端）：HashMap、RocksDB
         * 追求性能——HashMap：基于内存的（Flink推荐的）
         * 追求稳定性——RocksDB：基于磁盘的（性能会低很多，但是状态的大小取决于磁盘大小）
         *
         * 最终都会保存到HDFS
         *
         * 使用状态时一定要考虑状态的大小问题，不能让状态无限制的变大
         * 1、定期清理过期的状态
         * 2、尽量进行操作时控制状态的大小，例如：不做全局排名，只进行Top10排名
         *
         * savepoint：通过flink命令手动保存的CheckPoint
         */

        DataStream<String> lineDS = env.socketTextStream("master", 8888);

        // 统计每隔单词的数量
        // 数据格式：a,a,a,a,a

        // 1、将每一行数据的每个单词切开，并逐个发送到下游
        lineDS.process(new ProcessFunction<String, String>() {
                    @Override
                    // 每一条数据会执行一次
                    public void processElement(String line, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                        for (String word : line.split(",")) {
                            // 通过out将数据发送到下游
                            out.collect(word);
                        }
                    }
                })
                // 将数据按key进行分区
                .keyBy(w -> w, Types.STRING)
                // KeyBy之后也可以调用process方法，自定义处理逻辑
                // 需要接收三个泛型：keyBy的key的类型、DS中每条数据的类型、输出到下游的类型
                .process(new KeyedProcessFunction<String, String, String>() {
                    // Flink会自动给每个Key单独维护一个valueState
                    ValueState<Integer> valueState;

                    @Override
                    // 程序启动后只会执行一次
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("wordCntState", Types.INT));
                    }


                    @Override
                    // 每条数据会执行一次
                    // 有状态计算：每次计算时会考虑之前的计算结果（状态）
                    public void processElement(String word, KeyedProcessFunction<String, String, String>.Context ctx, Collector<String> out) throws Exception {
                        Integer lastCnt = valueState.value(); // 获取之前的结果
                        int currentCnt = 1;
                        if (lastCnt != null) { // 避免第一次取value时出现null
                            currentCnt = lastCnt + 1; // 计算新的结果
                        }
                        valueState.update(currentCnt); // 更新状态
                        out.collect(word + "," + currentCnt);
                    }
                }).print();


        env.execute();
    }
}
